package com.teemor.im.remoting.instance;

import com.teemor.im.common.SemaphoreReleaseOnlyOnce;
import com.teemor.im.common.command.RemoteCommand;
import com.teemor.im.common.exception.ImException;
import com.teemor.im.executor.CommandDispatcher;
import com.teemor.im.discovery.node.ServerNode;
import com.teemor.im.remoting.ReqCallback;
import com.teemor.im.remoting.exception.RemotingConnectClosedException;
import com.teemor.im.remoting.exception.RemotingSendRequestException;
import com.teemor.im.remoting.exception.RemotingTimeoutException;
import com.teemor.im.remoting.netty.ResponseFuture;
import io.netty.channel.Channel;
import io.netty.channel.ChannelFuture;
import lombok.extern.slf4j.Slf4j;

import java.util.Map;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Semaphore;
import java.util.concurrent.TimeUnit;

/**
 * @ename zhoulk
 * @cname 周李科
 * @date 2020/6/30 16:11
 */

@Slf4j
public abstract class NettyRemotingAbstract implements RemotingService {

    protected final Map<String, ResponseFuture> responseMap = new ConcurrentHashMap<>();


    //通过信号量进行限流，避免单向请求或者异步请求过多
    protected final Semaphore semaphoreOneway;
    protected final Semaphore semaphoreAsync;

    //获取回调接口执行线程池
    public abstract ExecutorService callbackExecutor();

    //获取与对应节点关联的Channel
    protected abstract Channel channel(ServerNode node);

    private CommandDispatcher dispatcher = new CommandDispatcher();

    /**
     * @param permitsOneway 单向请求允许最大量
     * @param permitsAsync  异步请求运行最大量
     */
    public NettyRemotingAbstract(final int permitsOneway, final int permitsAsync) {
        this.semaphoreOneway = new Semaphore(permitsOneway, true);
        this.semaphoreAsync = new Semaphore(permitsAsync, true);
    }

    @Override
    public ChannelFuture invokeRequest(RemoteCommand request, ServerNode node) throws RemotingConnectClosedException {
        Channel channel = this.channel(node);
        ChannelFuture future;
        if (channel != null && channel.isActive()) {
            try {
                future = channel.writeAndFlush(request);
            } catch (Exception e) {
                log.warn("send request exception, so close the node[{}]", node);
                throw new RemotingSendRequestException(node, e);
            }
        } else {
            throw new RemotingConnectClosedException(node);
        }
        return future;
    }

    @Override
    public RemoteCommand syncSend(RemoteCommand request, ServerNode target, long timeoutMillis) throws RemotingConnectClosedException, InterruptedException {
        String requestId = request.getCommandId();
        ResponseFuture responseFuture = new ResponseFuture(requestId, target, timeoutMillis, null, null);

        this.responseMap.put(requestId, responseFuture);
        try {
            ChannelFuture future = this.invokeRequest(request, target);
            future.addListener((f) -> {
                if (f.isSuccess()) {
                    responseFuture.setSendReqSuc(true);
                    return;
                } else {
                    responseFuture.setSendReqSuc(false);
                }
                responseMap.remove(requestId);
                responseFuture.setCause(f.cause());
                responseFuture.putResponse(null);
                log.warn("send a request command to node ip:{}, port:{} failed.", target.getHost(), target.getPort());
            });

            RemoteCommand responseCommand = responseFuture.waitResponse(timeoutMillis);
            if (null == responseCommand) {
                if (responseFuture.isSendReqSuc()) {
                    throw new RemotingTimeoutException(target, timeoutMillis,
                            responseFuture.getCause());
                } else {
                    throw new RemotingSendRequestException(target, responseFuture.getCause());
                }
            }
            return responseCommand;
        } catch (Exception e) {
            log.warn("send a request command to node ip:{}, port:{} failed.", target.getHost(), target.getPort());
            throw e;
        } finally {
            this.responseMap.remove(requestId);
        }
    }

    @Override
    public void asyncSend(RemoteCommand request, ServerNode target, ReqCallback callback, long timeoutMillis) throws InterruptedException {
        long beginStartTime = System.currentTimeMillis();
        String requestId = request.getCommandId();
        boolean acquired = this.semaphoreAsync.tryAcquire(timeoutMillis, TimeUnit.MILLISECONDS);
        if (acquired) {
            final SemaphoreReleaseOnlyOnce once = new SemaphoreReleaseOnlyOnce(this.semaphoreAsync);
            long costTime = System.currentTimeMillis() - beginStartTime;
            if (timeoutMillis < costTime) {
                once.release();
                throw new RemotingTimeoutException("invokeAsyncImpl call timeout");
            }
            ResponseFuture responseFuture = new ResponseFuture(requestId, target, timeoutMillis, callback, once);
            try {
                this.invokeRequest(request, target).addListener((f) -> {
                    if (f.isSuccess()) {
                        responseFuture.setSendReqSuc(true);
                    } else {
                        ResponseFuture rspFuture = responseMap.remove(requestId);
                        if (rspFuture != null) {
                            rspFuture.setSendReqSuc(false);
                            rspFuture.putResponse(null);
                            try {
                                executeInvokeCallback(rspFuture);
                            } catch (Throwable e) {
                                log.warn("execute callback in requestFail, and callback throw", e);
                            } finally {
                                rspFuture.release();
                            }
                        }
                    }
                });
            } catch (RemotingConnectClosedException e) {
                once.release();
            }
        } else {
            if (timeoutMillis <= 0) {
                throw new ImException("invokeAsyncImpl invoke too fast");
            } else {
                String info =
                        String.format("invokeAsyncImpl tryAcquire semaphore timeout, %dms, waiting thread nums: %d semaphoreAsyncValue: %d",
                                timeoutMillis,
                                this.semaphoreAsync.getQueueLength(),
                                this.semaphoreAsync.availablePermits()
                        );
                log.warn(info);
                throw new RemotingTimeoutException(info);
            }
        }
    }

    @Override
    public void oneWaySend(RemoteCommand request, ServerNode target, long timeoutMillis) throws InterruptedException {
        long beginStartTime = System.currentTimeMillis();
        boolean acquired = this.semaphoreOneway.tryAcquire(timeoutMillis, TimeUnit.MILLISECONDS);
        if (acquired) {
            final SemaphoreReleaseOnlyOnce once = new SemaphoreReleaseOnlyOnce(this.semaphoreOneway);
            long costTime = System.currentTimeMillis() - beginStartTime;
            if (timeoutMillis < costTime) {
                once.release();
                throw new RemotingTimeoutException("invokeOneWaySend call timeout");
            }
            try {
                this.invokeRequest(request, target).addListener((f) -> {
                    if (!f.isSuccess()) {
                        log.warn("send a request command to node <" + target.getHost() + ":" + target.getPort() + "> failed.");
                    }
                });
            } catch (RemotingConnectClosedException e) {
                once.release();
            }
        } else {
            if (timeoutMillis <= 0) {
                throw new ImException("invokeAsyncImpl invoke too fast");
            } else {
                String info =
                        String.format("invokeOneWaySend tryAcquire semaphore timeout, %dms, waiting thread nums: %d semaphoreOnewayValue: %d",
                                timeoutMillis,
                                this.semaphoreOneway.getQueueLength(),
                                this.semaphoreOneway.availablePermits()
                        );
                log.warn(info);
                throw new RemotingTimeoutException(info);
            }
        }
    }

    /**
     * 执行异步请求回调接口
     * @param future
     */
    private void executeInvokeCallback(ResponseFuture future) {
        ExecutorService callbackExecutor = this.callbackExecutor();
        boolean runInThisThread = false;
        if (callbackExecutor != null) {
            try {
                callbackExecutor.submit(() -> {
                    try {
                        future.executeCallback();
                    } catch (Exception e) {
                        log.warn("execute callback failed", e);
                    } finally {
                        future.release();
                    }
                });
            } catch (Exception e) {
                runInThisThread = true;
                log.warn("execute callback in executor exception, maybe executor busy", e);
            }
        } else {
            runInThisThread = true;
        }
        if (runInThisThread) {
            try {
                future.executeCallback();
            } catch (Exception e) {
                log.warn("execute callback failed", e);
            } finally {
                future.release();
            }
        }
    }
}
